/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.dubbo.registry.integration;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
import org.apache.dubbo.common.constants.RegistryConstants;
import org.apache.dubbo.common.deploy.ApplicationDeployer;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.common.url.component.ServiceConfigURL;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.common.utils.UrlUtils;
import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.registry.event.RegistryEvent;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.Registry;
import org.apache.dubbo.registry.RegistryFactory;
import org.apache.dubbo.registry.RegistryService;
import org.apache.dubbo.registry.client.ServiceDiscoveryRegistryDirectory;
import org.apache.dubbo.registry.client.migration.MigrationClusterInvoker;
import org.apache.dubbo.registry.client.migration.ServiceDiscoveryMigrationInvoker;
import org.apache.dubbo.registry.retry.ReExportTask;
import org.apache.dubbo.registry.support.SkipFailbackWrapperException;
import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.ProtocolServer;
import org.apache.dubbo.rpc.ProxyFactory;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Cluster;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
import org.apache.dubbo.rpc.cluster.Configurator;
import org.apache.dubbo.rpc.cluster.Constants;
import org.apache.dubbo.rpc.cluster.governance.GovernanceRuleRepository;
import org.apache.dubbo.rpc.cluster.support.MergeableCluster;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.ModuleModel;
import org.apache.dubbo.rpc.model.ProviderModel;
import org.apache.dubbo.rpc.model.ScopeModel;
import org.apache.dubbo.rpc.model.ScopeModelAware;
import org.apache.dubbo.rpc.model.ScopeModelUtil;
import org.apache.dubbo.rpc.protocol.InvokerWrapper;
import org.apache.dubbo.rpc.support.ProtocolUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.APPLICATION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.BACKGROUND_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.CLUSTER_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_VERSION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.ENABLED_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_MANAGEMENT_MODE;
import static org.apache.dubbo.common.constants.CommonConstants.EXTRA_KEYS_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.HIDE_KEY_PREFIX;
import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.IPV6_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.LOADBALANCE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.METHODS_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PACKABLE_METHOD_FACTORY_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PID_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.REGISTRY_LOCAL_FILE_CACHE_ENABLED;
import static org.apache.dubbo.common.constants.CommonConstants.REGISTRY_PROTOCOL_LISTENER_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.RELEASE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.SIDE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
import static org.apache.dubbo.common.constants.FilterConstants.VALIDATION_KEY;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_UNSUPPORTED_CATEGORY;
import static org.apache.dubbo.common.constants.QosConstants.ACCEPT_FOREIGN_IP;
import static org.apache.dubbo.common.constants.QosConstants.QOS_ENABLE;
import static org.apache.dubbo.common.constants.QosConstants.QOS_HOST;
import static org.apache.dubbo.common.constants.QosConstants.QOS_PORT;
import static org.apache.dubbo.common.constants.RegistryConstants.ALL_CATEGORIES;
import static org.apache.dubbo.common.constants.RegistryConstants.CATEGORY_KEY;
import static org.apache.dubbo.common.constants.RegistryConstants.CONFIGURATORS_CATEGORY;
import static org.apache.dubbo.common.constants.RegistryConstants.DYNAMIC_KEY;
import static org.apache.dubbo.common.constants.RegistryConstants.OVERRIDE_PROTOCOL;
import static org.apache.dubbo.common.constants.RegistryConstants.REGISTER_MODE_KEY;
import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_KEY;
import static org.apache.dubbo.common.constants.RegistryConstants.SERVICE_REGISTRY_PROTOCOL;
import static org.apache.dubbo.common.utils.StringUtils.isEmpty;
import static org.apache.dubbo.common.utils.UrlUtils.classifyUrls;
import static org.apache.dubbo.registry.Constants.CONFIGURATORS_SUFFIX;
import static org.apache.dubbo.registry.Constants.DEFAULT_REGISTRY_RETRY_PERIOD;
import static org.apache.dubbo.registry.Constants.ENABLE_26X_CONFIGURATION_LISTEN;
import static org.apache.dubbo.registry.Constants.ENABLE_CONFIGURATION_LISTEN;
import static org.apache.dubbo.registry.Constants.PROVIDER_PROTOCOL;
import static org.apache.dubbo.registry.Constants.REGISTER_IP_KEY;
import static org.apache.dubbo.registry.Constants.REGISTER_KEY;
import static org.apache.dubbo.registry.Constants.REGISTRY_RETRY_PERIOD_KEY;
import static org.apache.dubbo.registry.Constants.SIMPLIFIED_KEY;
import static org.apache.dubbo.remoting.Constants.BIND_IP_KEY;
import static org.apache.dubbo.remoting.Constants.BIND_PORT_KEY;
import static org.apache.dubbo.remoting.Constants.CHECK_KEY;
import static org.apache.dubbo.remoting.Constants.CODEC_KEY;
import static org.apache.dubbo.remoting.Constants.CONNECTIONS_KEY;
import static org.apache.dubbo.remoting.Constants.EXCHANGER_KEY;
import static org.apache.dubbo.remoting.Constants.PREFER_SERIALIZATION_KEY;
import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
import static org.apache.dubbo.rpc.Constants.DEPRECATED_KEY;
import static org.apache.dubbo.rpc.Constants.GENERIC_KEY;
import static org.apache.dubbo.rpc.Constants.INTERFACES;
import static org.apache.dubbo.rpc.Constants.MOCK_KEY;
import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.CONSUMER_URL_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.EXPORT_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.REFER_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.WARMUP_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.WEIGHT_KEY;
import static org.apache.dubbo.rpc.model.ScopeModelUtil.getApplicationModel;

/**
 * TODO, replace RegistryProtocol completely in the future.
 */
public class RegistryProtocol implements Protocol, ScopeModelAware {
    public static final String[] DEFAULT_REGISTER_PROVIDER_KEYS = {
        APPLICATION_KEY, CODEC_KEY, EXCHANGER_KEY, SERIALIZATION_KEY, PREFER_SERIALIZATION_KEY, CLUSTER_KEY, CONNECTIONS_KEY, DEPRECATED_KEY,
        GROUP_KEY, LOADBALANCE_KEY, MOCK_KEY, PATH_KEY, TIMEOUT_KEY, TOKEN_KEY, VERSION_KEY, WARMUP_KEY,
        WEIGHT_KEY, DUBBO_VERSION_KEY, RELEASE_KEY, SIDE_KEY, IPV6_KEY, PACKABLE_METHOD_FACTORY_KEY
    };

    public static final String[] DEFAULT_REGISTER_CONSUMER_KEYS = {
        APPLICATION_KEY, VERSION_KEY, GROUP_KEY, DUBBO_VERSION_KEY, RELEASE_KEY
    };

    private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(RegistryProtocol.class);

    private final Map<String, ServiceConfigurationListener> serviceConfigurationListeners = new ConcurrentHashMap<>();
    //To solve the problem of RMI repeated exposure port conflicts, the services that have been exposed are no longer exposed.
    //provider url <--> registry url <--> exporter
    private final Map<String, Map<String, ExporterChangeableWrapper<?>>> bounds = new ConcurrentHashMap<>();
    protected Protocol protocol;
    protected ProxyFactory proxyFactory;

    private ConcurrentMap<URL, ReExportTask> reExportFailedTasks = new ConcurrentHashMap<>();
    private HashedWheelTimer retryTimer = new HashedWheelTimer(new NamedThreadFactory("DubboReexportTimer", true), DEFAULT_REGISTRY_RETRY_PERIOD, TimeUnit.MILLISECONDS, 128);
    private FrameworkModel frameworkModel;
    private ExporterFactory exporterFactory;

    //Filter the parameters that do not need to be output in url(Starting with .)
    private static String[] getFilteredKeys(URL url) {
        Map<String, String> params = url.getParameters();
        if (CollectionUtils.isNotEmptyMap(params)) {
            return params.keySet().stream()
                .filter(k -> k.startsWith(HIDE_KEY_PREFIX))
                .toArray(String[]::new);
        } else {
            return new String[0];
        }
    }

    public RegistryProtocol() {
    }

    @Override
    public void setFrameworkModel(FrameworkModel frameworkModel) {
        this.frameworkModel = frameworkModel;
        this.exporterFactory = frameworkModel.getBeanFactory().getBean(ExporterFactory.class);
    }

    public void setProtocol(Protocol protocol) {
        this.protocol = protocol;
    }

    public void setProxyFactory(ProxyFactory proxyFactory) {
        this.proxyFactory = proxyFactory;
    }

    @Override
    public int getDefaultPort() {
        return 9090;
    }

    public Map<URL, Set<NotifyListener>> getOverrideListeners() {
        Map<URL, Set<NotifyListener>> map = new HashMap<>();
        List<ApplicationModel> applicationModels = frameworkModel.getApplicationModels();
        if (applicationModels.size() == 1) {
            return applicationModels.get(0).getBeanFactory().getBean(ProviderConfigurationListener.class).getOverrideListeners();
        } else {
            for (ApplicationModel applicationModel : applicationModels) {
                map.putAll(applicationModel.getBeanFactory().getBean(ProviderConfigurationListener.class).getOverrideListeners());
            }
        }
        return map;
    }

    private static void register(Registry registry, URL registeredProviderUrl) {
        ApplicationDeployer deployer = registeredProviderUrl.getOrDefaultApplicationModel().getDeployer();
        try {
            deployer.increaseServiceRefreshCount();

            String registryName = Optional.ofNullable(registry.getUrl())
                .map(u -> u.getParameter(RegistryConstants.REGISTRY_CLUSTER_KEY,
                    UrlUtils.isServiceDiscoveryURL(u) ? u.getParameter(REGISTRY_KEY) : u.getProtocol()))
                .filter(StringUtils::isNotEmpty)
                .orElse("unknown");

            MetricsEventBus.post(RegistryEvent.toRsEvent(registeredProviderUrl.getApplicationModel(), registeredProviderUrl.getServiceKey(), 1, Collections.singletonList(registryName)),
                () -> {
                // 服务注册
                    registry.register(registeredProviderUrl);
                    return null;
                });
        } finally {
            deployer.decreaseServiceRefreshCount();
        }
    }

    private void registerStatedUrl(URL registryUrl, URL registeredProviderUrl, boolean registered) {
        // 获取提供者模型对象
        ProviderModel model = (ProviderModel) registeredProviderUrl.getServiceModel();
        // 添加该服务注册状态的对象
        model.addStatedUrl(new ProviderModel.RegisterStatedURL(
            registeredProviderUrl,
            registryUrl,
            registered));
    }

    @Override
    public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
        // 获取注册中心地址
        URL registryUrl = getRegistryUrl(originInvoker);
        // url to export locally
        // 获取要暴露服务URL 对象
        URL providerUrl = getProviderUrl(originInvoker);

        // Subscribe the override data
        // FIXME When the provider subscribes, it will affect the scene : a certain JVM exposes the service and call
        //  the same service. Because the subscribed is cached key with the name of the service, it causes the
        //  subscription information to cover.

        /**
         * 获取 覆盖的url 对象 ，就是 把dubbo 换成 provider（具体用途不知道做什么）
         */
        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);

        // 创建覆盖监听器
        final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);

        // 获取 提供者 监听器Map
        Map<URL, Set<NotifyListener>> overrideListeners = getProviderConfigurationListener(overrideSubscribeUrl).getOverrideListeners();

        //  添加一个  监听器
        overrideListeners.computeIfAbsent(overrideSubscribeUrl, k -> new ConcurrentHashSet<>()).add(overrideSubscribeListener);

        /**
         * 覆盖 提供者的 暴露服务URL 对象
         * 这里是一个开发人员可以对暴露服务URL 对象进行修改的扩展配置，如：往URL对象添加什么参数。或者删除什么参数
         * 做到 定制配置 提供者的 暴露服务URL 对象
         */
        providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);

        //export invoker
        // 服务导出并启动netty 服务器 （核心步骤）
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker, providerUrl);


        // url to registry
        // 获取注册中心客服端
        final Registry registry = getRegistry(registryUrl);
        // 获取要注册的提供者地址
        final URL registeredProviderUrl = getUrlToRegistry(providerUrl, registryUrl);


        // decide if we need to delay publish (provider itself and registry should both need to register)
        // 获取该服务是否要注册到注册中心上
        boolean register = providerUrl.getParameter(REGISTER_KEY, true) && registryUrl.getParameter(REGISTER_KEY, true);
        // 判断是否注册到注册中心上
        if (register) {
            // 注册
            register(registry, registeredProviderUrl);
        }

        // register stated url on provider model
        // 服务的注册状态添加到提供者模型对象中
        registerStatedUrl(registryUrl, registeredProviderUrl, register);

        // 设置 服务提供者 URL 对象
        exporter.setRegisterUrl(registeredProviderUrl);
        // 服务订阅
        exporter.setSubscribeUrl(overrideSubscribeUrl);
        // 设置覆盖监听器
        exporter.setNotifyListener(overrideSubscribeListener);
        // 设置该服务是否已经注册
        exporter.setRegistered(register);

        ApplicationModel applicationModel = getApplicationModel(providerUrl.getScopeModel());

        if (applicationModel.modelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_26X_CONFIGURATION_LISTEN, true)) {

            if (!registry.isServiceDiscovery()) {

                // Deprecated! Subscribe to override rules in 2.6.x or before.
                registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
            }
        }

        notifyExport(exporter);
        //Ensure that a new exporter instance is returned every time export
        return new DestroyableExporter<>(exporter);
    }

    private <T> void notifyExport(ExporterChangeableWrapper<T> exporter) {
        ScopeModel scopeModel = exporter.getRegisterUrl().getScopeModel();
        // 获取注册协议的监听器
        List<RegistryProtocolListener> listeners = ScopeModelUtil.getExtensionLoader(RegistryProtocolListener.class, scopeModel).getActivateExtension(exporter.getOriginInvoker().getUrl(), REGISTRY_PROTOCOL_LISTENER_KEY);
        if (CollectionUtils.isNotEmpty(listeners)) {
            for (RegistryProtocolListener listener : listeners) {
                // 回调监听方法
                listener.onExport(this, exporter);
            }
        }
    }

    private URL overrideUrlWithConfig(URL providerUrl, OverrideListener listener) {
        // 获取提供者配置监听器对
        ProviderConfigurationListener providerConfigurationListener = getProviderConfigurationListener(providerUrl);
        // 覆盖 提供者的 暴露服务URL 对象
        providerUrl = providerConfigurationListener.overrideUrl(providerUrl);

        // 创建服务配置监听器 （吐槽：感觉dubbo 有些代码写的不怎么样，代码耦合度有点高，不像spring 源码一样 ）
        ServiceConfigurationListener serviceConfigurationListener = new ServiceConfigurationListener(providerUrl.getOrDefaultModuleModel(), providerUrl, listener);
        // 添加监听器
        serviceConfigurationListeners.put(providerUrl.getServiceKey(), serviceConfigurationListener);
        // 覆盖 提供者的 暴露服务URL 对象
        return serviceConfigurationListener.overrideUrl(providerUrl);
    }

    @SuppressWarnings("unchecked")
    private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker, URL providerUrl) {
        // 获取提供者的 地址 dubbo://192.168.31.130:20880/com.huan.dubbo.service.OrderService?anyhost=true&application=MyDubboProvider&background=false&bind.ip=192.168.31.130&bind.port=20880&deprecated=false&dubbo=2.0.2&executor-management-mode=isolation&file-cache=true&findOrder.return=true&findOrder.sent=true&generic=false&group=d1&interface=com.huan.dubbo.service.OrderService&methods=findOrder&pid=1724&prefer.serialization=fastjson2,hessian2&qos.port=22221&register-mode=instance&release=3.2.8-SNAPSHOT&revision=v1&service-name-mapping=true&side=provider&timestamp=1703948562553&version=v1
        String providerUrlKey = getProviderUrlKey(originInvoker);
        // 获取注册中心地址 service-discovery-registry://localdev:8848/org.apache.dubbo.registry.RegistryService?application=MyDubboProvider&dubbo=2.0.2&executor-management-mode=isolation&file-cache=true&pid=1724&qos.port=22221&register=false&register-mode=instance&registry=nacos&release=3.2.8-SNAPSHOT&timestamp=1703948557395
        String registryUrlKey = getRegistryUrlKey(originInvoker);
        // 把 服务提供者Invoker 对象封装一层
        Invoker<?> invokerDelegate = new InvokerDelegate<>(originInvoker, providerUrl);
        // 服务暴露 （重点核心）
        ReferenceCountExporter<?> exporter = exporterFactory.createExporter(providerUrlKey,
            // 服务暴露（核心）
            () -> protocol.export(invokerDelegate));


        return (ExporterChangeableWrapper<T>) bounds.computeIfAbsent(providerUrlKey, _k -> new ConcurrentHashMap<>())
            .computeIfAbsent(registryUrlKey, s -> {
                return new ExporterChangeableWrapper<>((ReferenceCountExporter<T>) exporter, originInvoker);
            });
    }

    public <T> void reExport(Exporter<T> exporter, URL newInvokerUrl) {
        if (exporter instanceof ExporterChangeableWrapper) {
            ExporterChangeableWrapper<T> exporterWrapper = (ExporterChangeableWrapper<T>) exporter;
            Invoker<T> originInvoker = exporterWrapper.getOriginInvoker();
            reExport(originInvoker, newInvokerUrl);
        }
    }

    /**
     * Reexport the invoker of the modified url
     *
     * @param originInvoker
     * @param newInvokerUrl
     * @param <T>
     */
    @SuppressWarnings("unchecked")
    public <T> void reExport(final Invoker<T> originInvoker, URL newInvokerUrl) {
        String providerUrlKey = getProviderUrlKey(originInvoker);
        String registryUrlKey = getRegistryUrlKey(originInvoker);
        Map<String, ExporterChangeableWrapper<?>> registryMap = bounds.get(providerUrlKey);
        if (registryMap == null) {
            logger.warn(INTERNAL_ERROR, "error state, exporterMap can not be null", "", "error state, exporterMap can not be null", new IllegalStateException("error state, exporterMap can not be null"));
            return;
        }
        ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) registryMap.get(registryUrlKey);
        if (exporter == null) {
            logger.warn(INTERNAL_ERROR, "error state, exporterMap can not be null", "", "error state, exporterMap can not be null", new IllegalStateException("error state, exporterMap can not be null"));
            return;
        }
        URL registeredUrl = exporter.getRegisterUrl();

        URL registryUrl = getRegistryUrl(originInvoker);
        URL newProviderUrl = getUrlToRegistry(newInvokerUrl, registryUrl);

        // update local exporter
        Invoker<T> invokerDelegate = new InvokerDelegate<T>(originInvoker, newInvokerUrl);
        exporter.setExporter(protocol.export(invokerDelegate));

        // update registry
        if (!newProviderUrl.equals(registeredUrl)) {
            try {
                doReExport(originInvoker, exporter, registryUrl, registeredUrl, newProviderUrl);
            } catch (Exception e) {
                ReExportTask oldTask = reExportFailedTasks.get(registeredUrl);
                if (oldTask != null) {
                    return;
                }
                ReExportTask task = new ReExportTask(
                    () -> doReExport(originInvoker, exporter, registryUrl, registeredUrl, newProviderUrl),
                    registeredUrl,
                    null
                );
                oldTask = reExportFailedTasks.putIfAbsent(registeredUrl, task);
                if (oldTask == null) {
                    // never has a retry task. then start a new task for retry.
                    retryTimer.newTimeout(task, registryUrl.getParameter(REGISTRY_RETRY_PERIOD_KEY, DEFAULT_REGISTRY_RETRY_PERIOD), TimeUnit.MILLISECONDS);
                }
            }
        }
    }

    private <T> void doReExport(final Invoker<T> originInvoker, ExporterChangeableWrapper<T> exporter,
                                URL registryUrl, URL oldProviderUrl, URL newProviderUrl) {
        if (exporter.isRegistered()) {
            Registry registry;
            try {
                registry = getRegistry(getRegistryUrl(originInvoker));
            } catch (Exception e) {
                throw new SkipFailbackWrapperException(e);
            }

            logger.info("Try to unregister old url: " + oldProviderUrl);
            registry.reExportUnregister(oldProviderUrl);

            logger.info("Try to register new url: " + newProviderUrl);
            registry.reExportRegister(newProviderUrl);
        }
        try {
            ProviderModel.RegisterStatedURL statedUrl = getStatedUrl(registryUrl, newProviderUrl);
            statedUrl.setProviderUrl(newProviderUrl);
            exporter.setRegisterUrl(newProviderUrl);
        } catch (Exception e) {
            throw new SkipFailbackWrapperException(e);
        }
    }

    private ProviderModel.RegisterStatedURL getStatedUrl(URL registryUrl, URL providerUrl) {
        ProviderModel providerModel = frameworkModel.getServiceRepository()
            .lookupExportedService(providerUrl.getServiceKey());

        List<ProviderModel.RegisterStatedURL> statedUrls = providerModel.getStatedUrl();
        return statedUrls.stream()
            .filter(u -> u.getRegistryUrl().equals(registryUrl)
                && u.getProviderUrl().getProtocol().equals(providerUrl.getProtocol()))
            .findFirst().orElseThrow(() -> new IllegalStateException("There should have at least one registered url."));
    }

    /**
     * Get an instance of registry based on the address of invoker
     *
     * @param registryUrl
     * @return
     */
    protected Registry getRegistry(final URL registryUrl) {
        // 获取注册中心工厂
        RegistryFactory registryFactory = ScopeModelUtil.getExtensionLoader(RegistryFactory.class, registryUrl.getScopeModel()).getAdaptiveExtension();
        // 获取注册中心
        return registryFactory.getRegistry(registryUrl);
    }

    protected URL getRegistryUrl(Invoker<?> originInvoker) {
        return originInvoker.getUrl();
    }

    protected URL getRegistryUrl(URL url) {
        if (SERVICE_REGISTRY_PROTOCOL.equals(url.getProtocol())) {
            return url;
        }
        return url.addParameter(REGISTRY_KEY, url.getProtocol()).setProtocol(SERVICE_REGISTRY_PROTOCOL);
    }

    /**
     * Return the url that is registered to the registry and filter the url parameter once
     *
     * @param providerUrl
     * @return url to registry.
     */
    private URL getUrlToRegistry(final URL providerUrl, final URL registryUrl) {
        //The address you see at the registry
        if (!registryUrl.getParameter(SIMPLIFIED_KEY, false)) {
            return providerUrl.removeParameters(getFilteredKeys(providerUrl)).removeParameters(
                MONITOR_KEY, BIND_IP_KEY, BIND_PORT_KEY, QOS_ENABLE, QOS_HOST, QOS_PORT, ACCEPT_FOREIGN_IP, VALIDATION_KEY,
                INTERFACES, REGISTER_MODE_KEY, PID_KEY, REGISTRY_LOCAL_FILE_CACHE_ENABLED, EXECUTOR_MANAGEMENT_MODE, BACKGROUND_KEY, ANYHOST_KEY);
        } else {
            String extraKeys = registryUrl.getParameter(EXTRA_KEYS_KEY, "");
            // if path is not the same as interface name then we should keep INTERFACE_KEY,
            // otherwise, the registry structure of zookeeper would be '/dubbo/path/providers',
            // but what we expect is '/dubbo/interface/providers'
            if (!providerUrl.getPath().equals(providerUrl.getParameter(INTERFACE_KEY))) {
                if (StringUtils.isNotEmpty(extraKeys)) {
                    extraKeys += ",";
                }
                extraKeys += INTERFACE_KEY;
            }
            String[] paramsToRegistry = getParamsToRegistry(DEFAULT_REGISTER_PROVIDER_KEYS
                , COMMA_SPLIT_PATTERN.split(extraKeys));
            return URL.valueOf(providerUrl, paramsToRegistry, providerUrl.getParameter(METHODS_KEY, (String[]) null));
        }

    }

    private URL getSubscribedOverrideUrl(URL registeredProviderUrl) {
        return registeredProviderUrl.setProtocol(PROVIDER_PROTOCOL)
            .addParameters(CATEGORY_KEY, CONFIGURATORS_CATEGORY, CHECK_KEY, String.valueOf(false));
    }

    /**
     * Get the address of the providerUrl through the url of the invoker
     *
     * @param originInvoker
     * @return
     */
    private URL getProviderUrl(final Invoker<?> originInvoker) {
        Object providerURL = originInvoker.getUrl().getAttribute(EXPORT_KEY);
        if (!(providerURL instanceof URL)) {
            throw new IllegalArgumentException("The registry export url is null! registry: " + originInvoker.getUrl().getAddress());
        }
        return (URL) providerURL;
    }

    /**
     * Get the key cached in bounds by invoker
     *
     * @param originInvoker
     * @return
     */
    private String getProviderUrlKey(final Invoker<?> originInvoker) {
        URL providerUrl = getProviderUrl(originInvoker);
        String key = providerUrl.removeParameters(DYNAMIC_KEY, ENABLED_KEY).toFullString();
        return key;
    }

    private String getRegistryUrlKey(final Invoker<?> originInvoker) {
        URL registryUrl = getRegistryUrl(originInvoker);
        String key = registryUrl.removeParameters(DYNAMIC_KEY, ENABLED_KEY).toFullString();
        return key;
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        // 从消费者Url 获取注册中心地址
        url = getRegistryUrl(url);

        // 获取注册中心客户端
        Registry registry = getRegistry(url);

        // 判断是否是 RegistryService 这个对象class
        if (RegistryService.class.equals(type)) {
            return proxyFactory.getInvoker((T) registry, type, url);
        }

        // group="a,b" or group="*"
        // 获取  @DubboReference(version = "3.0") 注解的 配置属性
        Map<String, String> qs = (Map<String, String>) url.getAttribute(REFER_KEY);
        // 获取组名
        String group = qs.get(GROUP_KEY);
        if (StringUtils.isNotEmpty(group)) {

            if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
                return doRefer(Cluster.getCluster(url.getScopeModel(), MergeableCluster.NAME), registry, type, url, qs);
            }
        }

        /**
         * 获取创建故障转移集群的 cluster 对象 ，这个对是用来创建 ClusterInvoker 对象的。重点方法是 doJoin 方法
         *
         * 调用集群容错 doInvoke （重点核心） 集群容错机制有如下几种：
         * 1. Failover Cluster （默认）
         *      @DubboReference(cluster = "failover",retries = 2)
         *      失败自动切换，当出现失败，重试其它服务器。通常用于读操作，但重试会带来更长延迟。可通过 retries="2" 来设置重试次数(不含第一次)。
         * 2. Failfast Cluster
         *      快速失败，只发起一次调用，失败立即报错。通常用于非幂等性的写操作，比如新增记录。
         *
         * 3. Failsafe Cluster
         *      失败安全，出现异常时，直接忽略。通常用于写入审计日志等操作。
         * 4. Failback Cluster
         *      失败自动恢复，后台记录失败请求，定时重发。通常用于消息通知操作。
         * 5. Forking Cluster
         *      并行调用多个服务器，只要一个成功即返回。通常用于实时性要求较高的读操作，但需要浪费更多服务资源。可通过 forks="2" 来设置最大并行数
         * 6. Broadcast Cluster
         *      广播调用所有提供者，逐个调用，任意一台报错则报错。通常用于通知所有提供者更新缓存或日志等本地资源信息。
         *      现在广播调用中，可以通过 broadcast.fail.percent 配置节点调用失败的比例，当达到这个比例后，BroadcastClusterInvoker 将不再调用其他节点，直接抛出异常。
         *      broadcast.fail.percent 取值在 0～100 范围内。默认情况下当全部调用失败后，才会抛出异常。
         *      broadcast.fail.percent 只是控制的当失败后是否继续调用其他节点，并不改变结果(任意一台报错则报错)。broadcast.fail.percent 参数 在 dubbo2.7.10 及以上版本生效。
         *      Broadcast Cluster 配置 broadcast.fail.percent。
         *      broadcast.fail.percent=20 代表了当 20% 的节点调用失败就抛出异常，不再调用其他节点。
         *
         *      @DubboReference(cluster = "broadcast", parameters = {"broadcast.fail.percent", "20"})
         * 7. Available Cluster
         *      调用目前可用的实例（只调用一个），如果当前没有可用的实例，则抛出异常。通常用于不需要负载均衡的场景。
         * 8. Mergeable Cluster
         *      将集群中的调用结果聚合起来返回结果，通常和group一起配合使用。通过分组对结果进行聚合并返回聚合后的结果，比如菜单服务，用group区分同一接口的多种实现，现在消费方需从每种group中调用一次并返回结果，对结果进行合并之后返回，这样就可以实现聚合菜单项。
         * 9. ZoneAware Cluster
         *      多注册中心订阅的场景，注册中心集群间的负载均衡。对于多注册中心间的选址策略有如下四种:
         *          1. 指定优先级：preferred="true"注册中心的地址将被优先选择
         *              <dubbo:registry address="zookeeper://127.0.0.1:2181" preferred="true" />
         *          2. 同中心优先：检查当前请求所属的区域，优先选择具有相同区域的注册中心
         *              <dubbo:registry address="zookeeper://127.0.0.1:2181" zone="beijing" />
         *          3. 权重轮询：根据每个注册中心的权重分配流量
         *              <dubbo:registry id="beijing" address="zookeeper://127.0.0.1:2181" weight="100" />
         *              <dubbo:registry id="shanghai" address="zookeeper://127.0.0.1:2182" weight="10" />
         */
        Cluster failoverCluster = Cluster.getCluster(url.getScopeModel(), qs.get(CLUSTER_KEY));

        // 远程服务引用（核心）
        return doRefer(failoverCluster, registry, type, url, qs);
    }

    protected <T> Invoker<T> doRefer(Cluster failoverCluster, Registry registry, Class<T> type, URL url, Map<String, String> parameters) {
        // 消费者属性
        Map<String, Object> consumerAttribute = new HashMap<>(url.getAttributes());

        // 移除 @DubboReference 这个注解的属性
        consumerAttribute.remove(REFER_KEY);

        String p = isEmpty(parameters.get(PROTOCOL_KEY)) ? CONSUMER : parameters.get(PROTOCOL_KEY);

        // 创建 消费者 Url 对象
        URL consumerUrl = new ServiceConfigURL(
            p,
            null,
            null,
            parameters.get(REGISTER_IP_KEY),
            0, getPath(parameters, type),
            parameters,
            consumerAttribute
        );

        // 往注册中心地址添加 消费者 URL 对象
        url = url.putAttribute(CONSUMER_URL_KEY, consumerUrl);

        //创建 一个 远程调用程序 MigrationInvoker  对象，此时对象里面的 Invoker 对象属性还没有值
        ClusterInvoker<T> migrationInvoker = getMigrationInvoker(this, failoverCluster, registry, type, url, consumerUrl);

        /**
         * 最后调用 RegistryProtocolListener 监听器 的 onRefer 方法 ，最后 把 MigrationInvoker 对象 返回
         * 注：此时对象里面的 Invoker 对象属性是有值了。这点是非常重要的 （这个步骤也是很重要理解的）
         */
        return interceptInvoker(migrationInvoker, url, consumerUrl);
    }

    private String getPath(Map<String, String> parameters, Class<?> type) {
        return !ProtocolUtils.isGeneric(parameters.get(GENERIC_KEY)) ? type.getName() : parameters.get(INTERFACE_KEY);
    }

    protected <T> ClusterInvoker<T> getMigrationInvoker(RegistryProtocol registryProtocol, Cluster cluster, Registry registry, Class<T> type, URL url, URL consumerUrl) {
        return new ServiceDiscoveryMigrationInvoker<T>(registryProtocol, cluster, registry, type, url, consumerUrl);
    }

    /**
     * This method tries to load all RegistryProtocolListener definitions, which are used to control the behaviour of invoker by interacting with defined, then uses those listeners to
     * change the status and behaviour of the MigrationInvoker.
     * <p>
     * Currently available Listener is MigrationRuleListener, one used to control the Migration behaviour with dynamically changing rules.
     *
     * @param invoker     MigrationInvoker that determines which type of invoker list to use
     * @param url         The original url generated during refer, more like a registry:// style url
     * @param consumerUrl Consumer url representing current interface and its config
     * @param <T>         The service definition
     * @return The @param MigrationInvoker passed in
     */
    protected <T> Invoker<T> interceptInvoker(ClusterInvoker<T> invoker, URL url, URL consumerUrl) {

        // 获取监听器
        List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);

        if (CollectionUtils.isEmpty(listeners)) {
            return invoker;
        }

        for (RegistryProtocolListener listener : listeners) {
            // 调用监听通知方法（核心重点步骤，服务注册到nacos 步骤就再里面）
            listener.onRefer(this, invoker, consumerUrl, url);
        }
        return invoker;
    }

    public <T> ClusterInvoker<T> getServiceDiscoveryInvoker(Cluster cluster, Registry registry, Class<T> type, URL url) {
        /**
         *     这里有涉及到dubbo 3.0 的 双注册模式配置：dubbo.application.register-mode=all
         *                       1. all 双注册 (默认)
         *                      2. instance 应用级注册
         *                      3. interface 接口级注册
         *
         *   这个对象是应用级别的服务发现
         */
        DynamicDirectory<T> directory = new ServiceDiscoveryRegistryDirectory<>(type, url);

        return doCreateInvoker(directory, cluster, registry, type);
    }

    public <T> ClusterInvoker<T> getInvoker(Cluster cluster, Registry registry, Class<T> type, URL url) {
        // FIXME, this method is currently not used, create the right registry before enable.
        /**
         *  创建  DynamicDirectory
         *     这里有涉及到dubbo 3.0 的 双注册模式配置：dubbo.application.register-mode=all
         *                       1. all 双注册 (默认)
         *                      2. instance 应用级注册
         *                      3. interface 接口级注册
         *
         *   这个对象是接口级别的服务发现
         */

        DynamicDirectory<T> directory = new RegistryDirectory<>(type, url);
        // 创建  Invoker （核心）
        return doCreateInvoker(directory, cluster, registry, type);
    }

    protected <T> ClusterInvoker<T> doCreateInvoker(DynamicDirectory<T> directory, Cluster cluster, Registry registry, Class<T> type) {
        // 设置注册中心连接客户端
        directory.setRegistry(registry);
        // 设置通信协议
        directory.setProtocol(protocol);
        // all attributes of REFER_KEY

        Map<String, String> parameters = new HashMap<>(directory.getConsumerUrl().getParameters());

        URL urlToRegistry = new ServiceConfigURL(
            parameters.get(PROTOCOL_KEY) == null ? CONSUMER : parameters.get(PROTOCOL_KEY),
            parameters.remove(REGISTER_IP_KEY),
            0,
            getPath(parameters, type),
            parameters
        );

        urlToRegistry = urlToRegistry.setScopeModel(directory.getConsumerUrl().getScopeModel());

        urlToRegistry = urlToRegistry.setServiceModel(directory.getConsumerUrl().getServiceModel());

        // 判断是否注册到注册中心上
        if (directory.isShouldRegister()) {
            // 设置 注册消费者 URL 对象
            directory.setRegisteredConsumerUrl(urlToRegistry);
            // 注册（核心之一）
            registry.register(directory.getRegisteredConsumerUrl());
        }
        /**
         * 构建路由链这个对象属性很重要，服务调用时候会从这个对象获取  Invoker 对象
         */
        directory.buildRouterChain(urlToRegistry);

        /**
         * 服务订阅 （重点核心之一）
         * 这里会连接注册中心获取提供者ip+端口号实例信息，然后把实例信息封装成 Invoker 对象放到内存中。
         * 再消费者调用的时候，不用再从注册中心获取了，而是本地缓存中获取！
         *
         * 这里面又设置到 这里有涉及到dubbo 3.0 的 双注册模式配置：dubbo.application.register-mode=all
         *
         *  RegistryDirectory 是接口级别的
         *  ServiceDiscoveryRegistryDirectory 是应用级别的
         */
        directory.subscribe(toSubscribeUrl(urlToRegistry));

        /**
         * 创建 获取故障转移集群的 ClusterInvoker 对象 ，这个对象很重要（重点核心）
         * 调用集群容错 doInvoke （重点核心） 集群容错机制有如下几种：
         * 1. Failover Cluster （默认）
         *      @DubboReference(cluster = "failover",retries = 2)
         *      失败自动切换，当出现失败，重试其它服务器。通常用于读操作，但重试会带来更长延迟。可通过 retries="2" 来设置重试次数(不含第一次)。
         * 2. Failfast Cluster
         *      快速失败，只发起一次调用，失败立即报错。通常用于非幂等性的写操作，比如新增记录。
         *
         * 3. Failsafe Cluster
         *      失败安全，出现异常时，直接忽略。通常用于写入审计日志等操作。
         * 4. Failback Cluster
         *      失败自动恢复，后台记录失败请求，定时重发。通常用于消息通知操作。
         * 5. Forking Cluster
         *      并行调用多个服务器，只要一个成功即返回。通常用于实时性要求较高的读操作，但需要浪费更多服务资源。可通过 forks="2" 来设置最大并行数
         * 6. Broadcast Cluster
         *      广播调用所有提供者，逐个调用，任意一台报错则报错。通常用于通知所有提供者更新缓存或日志等本地资源信息。
         *      现在广播调用中，可以通过 broadcast.fail.percent 配置节点调用失败的比例，当达到这个比例后，BroadcastClusterInvoker 将不再调用其他节点，直接抛出异常。
         *      broadcast.fail.percent 取值在 0～100 范围内。默认情况下当全部调用失败后，才会抛出异常。
         *      broadcast.fail.percent 只是控制的当失败后是否继续调用其他节点，并不改变结果(任意一台报错则报错)。broadcast.fail.percent 参数 在 dubbo2.7.10 及以上版本生效。
         *      Broadcast Cluster 配置 broadcast.fail.percent。
         *      broadcast.fail.percent=20 代表了当 20% 的节点调用失败就抛出异常，不再调用其他节点。
         *
         *      @DubboReference(cluster = "broadcast", parameters = {"broadcast.fail.percent", "20"})
         * 7. Available Cluster
         *      调用目前可用的实例（只调用一个），如果当前没有可用的实例，则抛出异常。通常用于不需要负载均衡的场景。
         * 8. Mergeable Cluster
         *      将集群中的调用结果聚合起来返回结果，通常和group一起配合使用。通过分组对结果进行聚合并返回聚合后的结果，比如菜单服务，用group区分同一接口的多种实现，现在消费方需从每种group中调用一次并返回结果，对结果进行合并之后返回，这样就可以实现聚合菜单项。
         * 9. ZoneAware Cluster
         *      多注册中心订阅的场景，注册中心集群间的负载均衡。对于多注册中心间的选址策略有如下四种:
         *          1. 指定优先级：preferred="true"注册中心的地址将被优先选择
         *              <dubbo:registry address="zookeeper://127.0.0.1:2181" preferred="true" />
         *          2. 同中心优先：检查当前请求所属的区域，优先选择具有相同区域的注册中心
         *              <dubbo:registry address="zookeeper://127.0.0.1:2181" zone="beijing" />
         *          3. 权重轮询：根据每个注册中心的权重分配流量
         *              <dubbo:registry id="beijing" address="zookeeper://127.0.0.1:2181" weight="100" />
         *              <dubbo:registry id="shanghai" address="zookeeper://127.0.0.1:2182" weight="10" />
         */
        return (ClusterInvoker<T>) cluster.join(directory, true);
    }

    public <T> void reRefer(ClusterInvoker<?> invoker, URL newSubscribeUrl) {
        if (!(invoker instanceof MigrationClusterInvoker)) {
            logger.error(REGISTRY_UNSUPPORTED_CATEGORY, "", "", "Only invoker type of MigrationClusterInvoker supports reRefer, current invoker is " + invoker.getClass());
            return;
        }

        MigrationClusterInvoker<?> migrationClusterInvoker = (MigrationClusterInvoker<?>) invoker;
        migrationClusterInvoker.reRefer(newSubscribeUrl);
    }

    public static URL toSubscribeUrl(URL url) {
        return url.addParameter(CATEGORY_KEY, ALL_CATEGORIES);
    }

    protected List<RegistryProtocolListener> findRegistryProtocolListeners(URL url) {
        return ScopeModelUtil.getExtensionLoader(RegistryProtocolListener.class, url.getScopeModel()).getActivateExtension(url, REGISTRY_PROTOCOL_LISTENER_KEY);
    }

    // available to test
    public String[] getParamsToRegistry(String[] defaultKeys, String[] additionalParameterKeys) {
        int additionalLen = additionalParameterKeys.length;
        String[] registryParams = new String[defaultKeys.length + additionalLen];
        System.arraycopy(defaultKeys, 0, registryParams, 0, defaultKeys.length);
        System.arraycopy(additionalParameterKeys, 0, registryParams, defaultKeys.length, additionalLen);
        return registryParams;
    }

    @Override
    public void destroy() {
        // FIXME all application models in framework are removed at this moment
        for (ApplicationModel applicationModel : frameworkModel.getApplicationModels()) {
            for (ModuleModel moduleModel : applicationModel.getModuleModels()) {
                List<RegistryProtocolListener> listeners = moduleModel.getExtensionLoader(RegistryProtocolListener.class)
                    .getLoadedExtensionInstances();
                if (CollectionUtils.isNotEmpty(listeners)) {
                    for (RegistryProtocolListener listener : listeners) {
                        listener.onDestroy();
                    }
                }
            }
        }

        for (ApplicationModel applicationModel : frameworkModel.getApplicationModels()) {
            if (applicationModel.modelEnvironment().getConfiguration().convert(Boolean.class, org.apache.dubbo.registry.Constants.ENABLE_CONFIGURATION_LISTEN, true)) {
                for (ModuleModel moduleModel : applicationModel.getPubModuleModels()) {
                    String applicationName = applicationModel.tryGetApplicationName();
                    if (applicationName == null) {
                        // already removed
                        continue;
                    }
                    if (moduleModel.getServiceRepository().getExportedServices().size() > 0) {
                        moduleModel.getExtensionLoader(GovernanceRuleRepository.class).getDefaultExtension()
                            .removeListener(applicationName + CONFIGURATORS_SUFFIX,
                                getProviderConfigurationListener(moduleModel));
                    }
                }
            }
        }

        List<Exporter<?>> exporters = bounds.values().stream().flatMap(e -> e.values().stream()).collect(Collectors.toList());
        for (Exporter<?> exporter : exporters) {
            exporter.unexport();
        }
        bounds.clear();
    }

    @Override
    public List<ProtocolServer> getServers() {
        return protocol.getServers();
    }

    //Merge the urls of configurators
    private static URL getConfiguredInvokerUrl(List<Configurator> configurators, URL url) {
        if (CollectionUtils.isNotEmpty(configurators)) {
            for (Configurator configurator : configurators) {
                // 定制配置 提供者的 暴露服务URL 对象
                url = configurator.configure(url);
            }
        }
        return url;
    }

    public static class InvokerDelegate<T> extends InvokerWrapper<T> {

        /**
         * @param invoker
         * @param url     invoker.getUrl return this value
         */
        public InvokerDelegate(Invoker<T> invoker, URL url) {
            super(invoker, url);
        }

        public Invoker<T> getInvoker() {
            if (invoker instanceof InvokerDelegate) {
                return ((InvokerDelegate<T>) invoker).getInvoker();
            } else {
                return invoker;
            }
        }
    }

    private static class DestroyableExporter<T> implements Exporter<T> {

        private Exporter<T> exporter;

        public DestroyableExporter(Exporter<T> exporter) {
            this.exporter = exporter;
        }

        @Override
        public Invoker<T> getInvoker() {
            return exporter.getInvoker();
        }

        @Override
        public void unexport() {
            exporter.unexport();
        }

        @Override
        public void register() {
            exporter.register();
        }

        @Override
        public void unregister() {
            exporter.unregister();
        }
    }

    /**
     * Reexport: the exporter destroy problem in protocol
     * 1.Ensure that the exporter returned by registry protocol can be normal destroyed
     * 2.No need to re-register to the registry after notify
     * 3.The invoker passed by the export method , would better to be the invoker of exporter
     */
    private class OverrideListener implements NotifyListener {
        private final URL subscribeUrl;
        private final Invoker originInvoker;

        private List<Configurator> configurators;

        public OverrideListener(URL subscribeUrl, Invoker originalInvoker) {
            this.subscribeUrl = subscribeUrl;
            this.originInvoker = originalInvoker;
        }

        /**
         * @param urls The list of registered information, is always not empty, The meaning is the same as the
         *             return value of {@link org.apache.dubbo.registry.RegistryService#lookup(URL)}.
         */
        @Override
        public synchronized void notify(List<URL> urls) {
            if (logger.isDebugEnabled()) {
                logger.debug("original override urls: " + urls);
            }

            List<URL> matchedUrls = getMatchedUrls(urls, subscribeUrl);
            if (logger.isDebugEnabled()) {
                logger.debug("subscribe url: " + subscribeUrl + ", override urls: " + matchedUrls);
            }

            // No matching results
            if (matchedUrls.isEmpty()) {
                return;
            }

            this.configurators = Configurator.toConfigurators(classifyUrls(matchedUrls, UrlUtils::isConfigurator))
                .orElse(configurators);

            ApplicationDeployer deployer = subscribeUrl.getOrDefaultApplicationModel().getDeployer();

            try {
                deployer.increaseServiceRefreshCount();
                doOverrideIfNecessary();
            } finally {
                deployer.decreaseServiceRefreshCount();
            }
        }

        public synchronized void doOverrideIfNecessary() {
            final Invoker<?> invoker;
            if (originInvoker instanceof InvokerDelegate) {
                invoker = ((InvokerDelegate<?>) originInvoker).getInvoker();
            } else {
                invoker = originInvoker;
            }
            //The origin invoker
            URL originUrl = RegistryProtocol.this.getProviderUrl(invoker);
            String providerUrlKey = getProviderUrlKey(originInvoker);
            String registryUrlKey = getRegistryUrlKey(originInvoker);
            Map<String, ExporterChangeableWrapper<?>> exporterMap = bounds.get(providerUrlKey);
            if (exporterMap == null) {
                logger.warn(INTERNAL_ERROR, "error state, exporterMap can not be null", "", "error state, exporterMap can not be null", new IllegalStateException("error state, exporterMap can not be null"));
                return;
            }
            ExporterChangeableWrapper<?> exporter = exporterMap.get(registryUrlKey);
            if (exporter == null) {
                logger.warn(INTERNAL_ERROR, "unknown error in registry module", "", "error state, exporter should not be null", new IllegalStateException("error state, exporter should not be null"));
                return;
            }
            //The current, may have been merged many times
            Invoker<?> exporterInvoker = exporter.getInvoker();
            URL currentUrl = exporterInvoker == null ? null : exporterInvoker.getUrl();
            //Merged with this configuration
            URL newUrl = getConfiguredInvokerUrl(configurators, originUrl);
            newUrl = getConfiguredInvokerUrl(getProviderConfigurationListener(originUrl).getConfigurators(), newUrl);
            newUrl = getConfiguredInvokerUrl(serviceConfigurationListeners.get(originUrl.getServiceKey())
                .getConfigurators(), newUrl);
            if (!newUrl.equals(currentUrl)) {
                if (newUrl.getParameter(Constants.NEED_REEXPORT, true)) {
                    RegistryProtocol.this.reExport(originInvoker, newUrl);
                }
                logger.info("exported provider url changed, origin url: " + originUrl +
                    ", old export url: " + currentUrl + ", new export url: " + newUrl);
            }
        }

        private List<URL> getMatchedUrls(List<URL> configuratorUrls, URL currentSubscribe) {
            List<URL> result = new ArrayList<>();
            for (URL url : configuratorUrls) {
                URL overrideUrl = url;
                // Compatible with the old version
                if (url.getCategory() == null && OVERRIDE_PROTOCOL.equals(url.getProtocol())) {
                    overrideUrl = url.addParameter(CATEGORY_KEY, CONFIGURATORS_CATEGORY);
                }

                // Check whether url is to be applied to the current service
                if (UrlUtils.isMatch(currentSubscribe, overrideUrl)) {
                    result.add(url);
                }
            }
            return result;
        }
    }

    private ProviderConfigurationListener getProviderConfigurationListener(URL url) {
        // 获取提供者配置监听器
        return getProviderConfigurationListener(url.getOrDefaultModuleModel());
    }

    private ProviderConfigurationListener getProviderConfigurationListener(ModuleModel moduleModel) {
        // 获取提供者配置监听器如果spring ioc 容器没有的 ，就注册 ProviderConfigurationListener 到spring ioc 容器中
        return moduleModel.getBeanFactory().getOrRegisterBean(ProviderConfigurationListener.class, type -> new ProviderConfigurationListener(moduleModel));
    }

    private class ServiceConfigurationListener extends AbstractConfiguratorListener {
        private URL providerUrl;
        private OverrideListener notifyListener;

        private final ModuleModel moduleModel;

        public ServiceConfigurationListener(ModuleModel moduleModel, URL providerUrl, OverrideListener notifyListener) {
            super(moduleModel);
            this.providerUrl = providerUrl;
            this.notifyListener = notifyListener;
            this.moduleModel = moduleModel;
            if (moduleModel.modelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_CONFIGURATION_LISTEN, true)) {
                this.initWith(DynamicConfiguration.getRuleKey(providerUrl) + CONFIGURATORS_SUFFIX);
            }
        }

        private <T> URL overrideUrl(URL providerUrl) {
            return RegistryProtocol.getConfiguredInvokerUrl(configurators, providerUrl);
        }

        @Override
        protected void notifyOverrides() {
            ApplicationDeployer deployer = this.moduleModel.getApplicationModel().getDeployer();
            try {
                deployer.increaseServiceRefreshCount();
                notifyListener.doOverrideIfNecessary();
            } finally {
                deployer.decreaseServiceRefreshCount();
            }
        }
    }

    private class ProviderConfigurationListener extends AbstractConfiguratorListener {

        private final Map<URL, Set<NotifyListener>> overrideListeners = new ConcurrentHashMap<>();

        private final ModuleModel moduleModel;

        public ProviderConfigurationListener(ModuleModel moduleModel) {
            super(moduleModel);
            this.moduleModel = moduleModel;

            if (moduleModel.modelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_CONFIGURATION_LISTEN, true)) {
                this.initWith(moduleModel.getApplicationModel().getApplicationName() + CONFIGURATORS_SUFFIX);
            }
        }

        /**
         * Get existing configuration rule and override provider url before exporting.
         *
         * @param providerUrl
         * @param <T>
         * @return
         */
        private <T> URL overrideUrl(URL providerUrl) {
            return RegistryProtocol.getConfiguredInvokerUrl(configurators, providerUrl);
        }

        @Override
        protected void notifyOverrides() {
            ApplicationDeployer deployer = this.moduleModel.getApplicationModel().getDeployer();
            try {
                deployer.increaseServiceRefreshCount();
                overrideListeners.values().forEach(listeners -> {
                    for (NotifyListener listener : listeners) {
                        ((OverrideListener) listener).doOverrideIfNecessary();
                    }
                });
            } finally {
                deployer.decreaseServiceRefreshCount();
            }
        }

        public Map<URL, Set<NotifyListener>> getOverrideListeners() {
            return overrideListeners;
        }
    }

    /**
     * exporter proxy, establish the corresponding relationship between the returned exporter and the exporter
     * exported by the protocol, and can modify the relationship at the time of override.
     *
     * @param <T>
     */
    private class ExporterChangeableWrapper<T> implements Exporter<T> {

        private final ScheduledExecutorService executor;

        private final Invoker<T> originInvoker;
        private Exporter<T> exporter;
        private URL subscribeUrl;
        private URL registerUrl;

        private NotifyListener notifyListener;
        private final AtomicBoolean registered = new AtomicBoolean(false);

        public ExporterChangeableWrapper(ReferenceCountExporter<T> exporter, Invoker<T> originInvoker) {
            this.exporter = exporter;
            exporter.increaseCount();
            this.originInvoker = originInvoker;
            FrameworkExecutorRepository frameworkExecutorRepository = originInvoker.getUrl().getOrDefaultFrameworkModel().getBeanFactory()
                .getBean(FrameworkExecutorRepository.class);
            this.executor = frameworkExecutorRepository.getSharedScheduledExecutor();
        }

        public Invoker<T> getOriginInvoker() {
            return originInvoker;
        }

        @Override
        public Invoker<T> getInvoker() {
            return exporter.getInvoker();
        }

        public void setExporter(Exporter<T> exporter) {
            this.exporter = exporter;
        }

        @Override
        public void register() {
            if (registered.compareAndSet(false, true)) {
                // 获取注册地址
                URL registryUrl = getRegistryUrl(originInvoker);
                // 获取注册中心客户端
                Registry registry = getRegistry(registryUrl);
                // 服务注册到注册中心（核心步骤）
                RegistryProtocol.register(registry, getRegisterUrl());

                ProviderModel providerModel = frameworkModel.getServiceRepository().lookupExportedService(getRegisterUrl().getServiceKey());

                List<ProviderModel.RegisterStatedURL> statedUrls = providerModel.getStatedUrl();

                statedUrls.stream().filter(u -> u.getRegistryUrl().equals(registryUrl) && u.getProviderUrl().getProtocol().equals(getRegisterUrl().getProtocol()))
                    .forEach(u -> u.setRegistered(true));
                logger.info("Registered dubbo service " + getRegisterUrl().getServiceKey() + " url " + getRegisterUrl() + " to registry " + registryUrl);
            }
        }

        @Override
        public synchronized void unregister() {
            if (registered.compareAndSet(true, false)) {
                URL registryUrl = getRegistryUrl(originInvoker);
                Registry registry = RegistryProtocol.this.getRegistry(registryUrl);

                ProviderModel providerModel = frameworkModel.getServiceRepository()
                    .lookupExportedService(getRegisterUrl().getServiceKey());

                List<ProviderModel.RegisterStatedURL> statedURLs =
                    providerModel.getStatedUrl()
                        .stream()
                        .filter(u -> u.getRegistryUrl().equals(registryUrl)
                            && u.getProviderUrl().getProtocol().equals(getRegisterUrl().getProtocol()))
                        .collect(Collectors.toList());
                if (statedURLs.isEmpty() || statedURLs.stream().anyMatch(ProviderModel.RegisterStatedURL::isRegistered)) {
                    try {
                        registry.unregister(registerUrl);
                    } catch (Throwable t) {
                        logger.warn(INTERNAL_ERROR, "unknown error in registry module", "", t.getMessage(), t);
                    }
                }

                try {
                    if (subscribeUrl != null) {
                        Map<URL, Set<NotifyListener>> overrideListeners = getProviderConfigurationListener(subscribeUrl).getOverrideListeners();
                        Set<NotifyListener> listeners = overrideListeners.get(subscribeUrl);
                        if (listeners != null) {
                            if (listeners.remove(notifyListener)) {
                                ApplicationModel applicationModel = getApplicationModel(registerUrl.getScopeModel());
                                if (applicationModel.modelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_26X_CONFIGURATION_LISTEN, true)) {
                                    if (!registry.isServiceDiscovery()) {
                                        registry.unsubscribe(subscribeUrl, notifyListener);
                                    }
                                }
                                if (applicationModel.modelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_CONFIGURATION_LISTEN, true)) {
                                    for (ModuleModel moduleModel : applicationModel.getPubModuleModels()) {
                                        if (moduleModel.getServiceRepository().getExportedServices().size() > 0) {
                                            moduleModel.getExtensionLoader(GovernanceRuleRepository.class).getDefaultExtension()
                                                .removeListener(subscribeUrl.getServiceKey() + CONFIGURATORS_SUFFIX,
                                                    serviceConfigurationListeners.remove(subscribeUrl.getServiceKey()));
                                        }
                                    }
                                }
                            }
                            if (listeners.isEmpty()) {
                                overrideListeners.remove(subscribeUrl);
                            }
                        }
                    }
                } catch (Throwable t) {
                    logger.warn(INTERNAL_ERROR, "unknown error in registry module", "", t.getMessage(), t);
                }
            }
        }

        @Override
        public synchronized void unexport() {
            String providerUrlKey = getProviderUrlKey(this.originInvoker);
            String registryUrlKey = getRegistryUrlKey(this.originInvoker);
            Map<String, ExporterChangeableWrapper<?>> exporterMap = bounds.remove(providerUrlKey);
            if (exporterMap != null) {
                exporterMap.remove(registryUrlKey);
            }

            unregister();
            doUnExport();
        }

        public void setRegistered(boolean registered) {
            this.registered.set(registered);
        }

        public boolean isRegistered() {
            return registered.get();
        }

        private void doUnExport() {
            try {
                exporter.unexport();
            } catch (Throwable t) {
                logger.warn(INTERNAL_ERROR, "unknown error in registry module", "", t.getMessage(), t);
            }
        }

        public void setSubscribeUrl(URL subscribeUrl) {
            this.subscribeUrl = subscribeUrl;
        }

        public void setRegisterUrl(URL registerUrl) {
            this.registerUrl = registerUrl;
        }

        public void setNotifyListener(NotifyListener notifyListener) {
            this.notifyListener = notifyListener;
        }

        public URL getRegisterUrl() {
            return registerUrl;
        }
    }

}
